package com.kafka;

import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.example.ProductCreatedEvent;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;

/**
 * @author Administrator
 * 优先级：java config > application.properties
 */
@Configuration
public class KafkaProducerConfig {
    @Value("${spring.kafka.producer.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.producer.key-serializer}")
    private String keySerializer;
    @Value("${spring.kafka.producer.value-serializer}")
    private String valueSerializer;
    @Value("${spring.kafka.producer.acks}")
    private String acks;
    @Value("${spring.kafka.producer.properties.delivery.timeout.ms}")
    private String deliveryTimeout;

    @Value("${spring.kafka.producer.properties.linger.ms}")
    private String linger;
    @Value("${spring.kafka.producer.properties.request.timeout.ms}")
    private String requestTimeout;
    @Value("${spring.kafka.producer.properties.enable.idempotence}")
    private boolean idempotence;

    @Value("${spring.kafka.producer.properties.max.in.flight.requests.per.connection}")
    private Integer inflightConnection;
    Map<String,Object> producerConfigs(){
        Map<String,Object>config=new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrapServers);
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,keySerializer);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,valueSerializer);
        //要想幂等性起作用 acks必须是all
        config.put(ProducerConfig.ACKS_CONFIG,acks);
        //要想幂等性起作用 重试次数必须大于0 由于我们使用了 DELIVERY_TIMEOUT_MS_CONFIG，LINGER_MS_CONFIG，REQUEST_TIMEOUT_MS_CONFIG这三个配置
        //这个配置就可以删除了或者注释掉 kafka会默认用默认值
      // config.put(ProducerConfig.RETRIES_CONFIG,Integer.MAX_VALUE);
        config.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG,deliveryTimeout);
        config.put(ProducerConfig.LINGER_MS_CONFIG,linger);
        config.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,requestTimeout);
        //开启幂等 MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION  ACKS_CONFIG RETRIES_CONFIG 这三个配置值不能和幂等性产生冲突
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,idempotence);

        /*
         * 客户端在阻塞之前将在单个连接上发送的最大未确认请求数。
         * 请注意，如果将此配置设置为大于 1 且 <code>enable.idempotence</code> 设置为 false，则由于重试（即，如果启用了重试），
         * 发送失败后存在消息重新排序的风险；如果禁用重试或将 <code>enable.idempotence</code> 设置为 true，则将保留排序。
         * 此外，启用幂等性要求此配置的值小于或等于 5。
         * 如果设置了冲突的配置并且未明确启用幂等性，则幂等性将被禁用。
         * 这个配置项限制了每个连接上未完成的请求的最大数量。未完成的请求是指那些已经被发送但还没有收到响应的请求。
            影响顺序性和可靠性：设置这个值会影响消息的顺序性和可靠性。
         */
        //如果要想幂等性起作用 该值不能大于5
         config.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION,inflightConnection);
        return config;
    }
    @Bean
    ProducerFactory<String, ProductCreatedEvent>producerFactory(){
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }
    @Bean
    KafkaTemplate<String,ProductCreatedEvent>kafkaTemplate(){
        return new KafkaTemplate<String,ProductCreatedEvent>(producerFactory());
    }

    @Bean
    public KafkaAdmin kafkaAdmin() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        return new KafkaAdmin(configs);
    }
    @Bean
    NewTopic createTopic(){
        return TopicBuilder.name("huran-test")
                .partitions(1)
                .replicas(3)
                .configs(Map.of("min.insync.replicas","2"))
                .build();
    }
}
